package com.ecreditpal.audio.netty.handler;


import com.ecreditpal.audio.WorkDispatcher;
import com.ecreditpal.audio.model.AudioModel;
import com.ecreditpal.audio.model.AudioSession;
import com.ecreditpal.audio.platform.PlatformBalancer;
import com.ecreditpal.audio.rpc.client.RpcSpeechClientService;
import com.ecreditpal.audio.service.SeechRecordService;
import com.ecreditpal.audio.util.SpringContextUtil;
import com.google.protobuf.MessageLite;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author lifeng
 * @version 1.0 on 2018/7/3.
 */
@Slf4j
public class SubReqServerHandler extends ChannelInboundHandlerAdapter {

    private final int MAX_OVERTIME  = 3;
    private Map<ChannelHandlerContext,Integer> clientOvertimeMap = new ConcurrentHashMap<>();


    private RpcSpeechClientService speechService;
    private SeechRecordService speechRecordService;
    private PlatformBalancer platformBalancer;
    private AudioSession audioSession;
    private int byteRate = 16000 * 16 / 8;


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }
//    private LinkedBlockingQueue<AudioModel.Audio> queue = new LinkedBlockingQueue<>();


    public SubReqServerHandler() {
        this.speechService = SpringContextUtil.getBean(RpcSpeechClientService.class);
        this.platformBalancer = SpringContextUtil.getBean(PlatformBalancer.class);
        this.speechRecordService = SpringContextUtil.getBean(SeechRecordService.class);
        this.audioSession = new AudioSession(byteRate);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException, InterruptedException {
        if (!(msg instanceof MessageLite)) {
            ctx.fireChannelRead(msg);
            return;
        }
        AudioModel.Audio req = (AudioModel.Audio) msg;
        audioSession.getQueue().put(req);
        log.info("index {}", req.getIndex());

        if (audioSession.isStarted()) {

        } else {
            audioSession.setStarted(true);
            AudioProcessor processor = new AudioProcessor(audioSession, speechService, speechRecordService,ctx, platformBalancer);
            WorkDispatcher.getInstance().taskExecute(processor);
        }


//        Runnable runnable = new Runnable() {
//            @Override
//            public void run() {
//                while (true) {
//                    AudioModel.Audio req = queue.poll();
//                    if (req == null) {
//                        log.info("req is null");
//                        continue;
//                    } else {
//                        log.info("task index {}",req.getIndex());
//                    }
//
//                    ByteBuffer bf = req.getData().asReadOnlyByteBuffer();
//                    log.info("get buffer");
//                    if (bf.hasRemaining()) {
//                        byte[] newMsg = new byte[bf.remaining()];
//                        bf.get(newMsg);
//                        log.info("get buffer success,length:{}",newMsg.length);
//
//
//                        byte[] data;
//                        if (req.getType() == AudioModel.AudioType.LOCAL) {
//
//                            localLeft = localLeft == null ? newMsg :
//                                    BaseUtil.mergeBytes(localLeft, newMsg);
//                            data = localLeft;
//                        } else {
//                            remoteLeft = remoteLeft == null ? newMsg :
//                                    BaseUtil.mergeBytes(remoteLeft, newMsg);
//                            data = remoteLeft;
//                        }
//
//                        List<Double> points = null;
//                        points = speechService.sendMessage(data);
//
//
//                        log.info("cut size: {}", points.size());
//                        double offset = 0.0;
//                        if (points.size() > 0) {
//                            for (Double p : points) {
//                                AudioModel.Text.Builder builder = AudioModel.Text.newBuilder();
//                                builder.setCaseid(req.getCaseid());
//                                builder.setType(req.getType());
//                                builder.setBeginTime((int) (totalOffset == 0 ? offset : totalOffset + offset) * 1000);
//                                builder.setEndTime((int) ((totalOffset + p) * 1000));
//
//                                log.info("start:{},end:{}", offset * 1000, (totalOffset + p) * 1000);
//
//                                AsrCallback asrCallback = new AsrCallback(ctx, builder);
//                                byte[] chunk = Wav.cutBody(data, offset, p, byteRate);
//
//                                AsrRunner asrRunner;
//                                if (isLongChunk) {
//                                    asrRunner = platformBalancer.getLongRunner();
//                                    isLongChunk = false;
//                                } else if (p - offset > 5) {
//                                    asrRunner = platformBalancer.getLongRunner();
//                                } else {
//                                    asrRunner = platformBalancer.chooseRunner();
//                                }
//                                asrRunner.process(chunk, asrCallback);
//
//                                //更新最近一个分割的节点
//                                offset = p;
//                            }
//                            totalOffset += offset;
//                            if (req.getType() == AudioModel.AudioType.LOCAL) {
//                                localLeft = Wav.cutSingle(localLeft, offset, byteRate);
//                            } else {
//                                remoteLeft = Wav.cutSingle(remoteLeft, offset, byteRate);
//                            }
//                        } else {
//                            isLongChunk = true;
//                        }
////                            else {
////                                AudioModel.Text.Builder builder = AudioModel.Text.newBuilder();
////                                builder.setCaseid(req.getCaseid());
////                                builder.setType(req.getType());
////                                builder.setBeginTime((int) (totalOffset * 1000));
////                                builder.setEndTime((int) ((totalOffset = totalOffset + Wav.getWavLengthNoBody(localLeft, simpleRate)) * 1000));
////
////                                AsrCallback asrCallback = new AsrCallback(ctx, builder);
////                                AsrRunner asrRunner = platformBalancer.getLongRunner();
////                                asrRunner.process(localLeft, asrCallback);
////                                localLeft = null;
////                            }
//
//                        if (req.getIsLast() && req.getType() == AudioModel.AudioType.LOCAL) {
//                            log.info("it is the last chunk,index: {}",req.getIndex());
//                            AsrRunner asrRunner = isLongChunk ? platformBalancer.getLongRunner() : platformBalancer.chooseRunner();
//                            AudioModel.Text.Builder builder = AudioModel.Text.newBuilder();
//                            builder.setCaseid(req.getCaseid());
//                            builder.setType(req.getType());
//                            builder.setBeginTime((int) (totalOffset * 1000));
//                            builder.setEndTime((int) ((totalOffset = totalOffset + Wav.getWavLengthNoBody(localLeft, byteRate)) * 1000));
//                            AsrCallback asrCallback = new AsrCallback(ctx, builder);
//                            asrRunner.process(localLeft, asrCallback);
//                            isStarted  = false;
//                        }
//                    }
//                }
//            }
//        };
//        WorkDispatcher.getInstance().taskExecute(processor);
    }


//    @Override
//    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//        log.error("can not get data", cause);
//
//    }

    public static void main(String[] args) {
        ExecutorService e =Executors.newFixedThreadPool(5);
        e.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println(123);
            }
        });
        e.shutdown();
    }



}
